package com.qiguliuxing.dts.core.redis;

import com.alibaba.fastjson.JSON;
import com.qiguliuxing.dts.core.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.ParameterizedType;
import java.util.concurrent.TimeUnit;

public abstract class RedisListener<T> implements MessageListener {

    @Autowired
    protected RedisTemplate<String, Object> redisTemplate;

    @Autowired
    protected RedisMessageListenerContainer messageListenerContainer;

    @Autowired
    protected TopicSubscriber subscriber;

    @Override
    public void onMessage(org.springframework.data.redis.connection.Message message, byte[] bytes) {
        String name = this.getClass().getSimpleName();
        String topic = new String(message.getChannel());
        String content = new String(message.getBody());
        Message m = JSON.parseObject(content, Message.class);
        String key = String.join("_", topic, name, m.getId());

        Object b = redisTemplate.opsForList().rightPop(topic + "_" + name);

        if (b != null && b.equals(m.getTopic())) {
            T t = JSON.parseObject(m.getContent(),((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]);
            handler(t);// 处理redis消息。
            // set data expire.使用redis的expire接口直接丢弃消费过的数据。
            redisTemplate.expire(key, 1, TimeUnit.NANOSECONDS);
        } else {
            // todo retry
            redisTemplate.opsForValue().set("fail_" + key, content);
        }
    }

    protected abstract void handler(T t);
}
